package MLlib2

import org.apache.spark.ml.linalg
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.ml.regression.{LinearRegression, LinearRegressionModel}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import util.SparkUtil

import scala.collection.mutable

/**
 * 线性回归算法调用示例
 * 寻找房价和面积、楼层之间的线性函数关系
 */
object EasyLinearRegressionDemo {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkUtil.getSparkSession(this.getClass.getSimpleName)
    import org.apache.spark.sql.functions._
    import spark.implicits._
    // inferSchema自动推断它的schema信息（Double）
    val sample: DataFrame = spark.read.option("header", value = true).option("inferSchema", value = true).csv("userprofile/data/Demo/house/h.csv")
    //    sample.printSchema()

    //    val arr2vec: UserDefinedFunction = udf((arr: mutable.WrappedArray[Double]) => {
    //      Vectors.dense(arr.toArray)
    //    })
    //    val frame: DataFrame = sample.select(arr2vec(array('area, 'floor)).as("vec"), 'price.as("label"))   // sql方式向量化
    val vecSample: DataFrame = sample.map({
      case Row(area: Double, floor: Double, price: Double) =>
        val features: linalg.Vector = Vectors.dense(Array(area, floor))
        (features, price)
    }).toDF("features", "label")    // rdd方式向量化
    vecSample.show(100)

    // 构造算法工具
    val regression: LinearRegression = new LinearRegression()
      .setFeaturesCol("features")
      .setLabelCol("label")
      .setRegParam(0.01)    // 正则化参数，用于防止过拟合；样本中离群点越多，这个参数可以调的越大

    val model: LinearRegressionModel = regression.fit(vecSample)
    val test: DataFrame = spark.read.option("header", value = true).option("inferSchema", value = true).csv("userprofile/data/Demo/house/t.csv")
    val vecTest: DataFrame = test.rdd.map({
      case Row(id: Int, area: Double, floor: Double) =>
        val features: linalg.Vector = Vectors.dense(Array(area, floor))
        (id, features)
    }).toDF("id", "features")

    val res0: DataFrame = model.transform(vecSample)
    res0.show(100)
    val res: DataFrame = model.transform(vecTest)
    res.show(100)

  }
}
